package source;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;

/**
 * flink kafka source
 */
public class KafKaSource {

    public static void main(String[] args) throws Exception{
        // 创建flink本地执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        // 配置
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers","master:9092,slave1:9092,slave2:9092");
        properties.setProperty("group.id","test");
        properties.setProperty("auto.offset.reset","earliest");
        // 添加kafka source
        DataStreamSource<String> stream = env.addSource(new FlinkKafkaConsumer<String>("wc", new SimpleStringSchema(), properties));

        // source 并行度
        int parallelism = stream.getParallelism();
        System.out.println("source parallelism: " + parallelism);

        // map
        SingleOutputStreamOperator<Tuple2<String, String>> mapped = stream.map(new MapFunction<String, Tuple2<String, String>>() {
            @Override
            public Tuple2<String, String> map(String s) throws Exception {
                String[] split = s.split(",");
                return Tuple2.of(split[0], split[1]);
            }
        });
        // transformation 并行度
        System.out.println("transformation parallelism: " + mapped.getParallelism());

        // sink
        DataStreamSink<Tuple2<String, String>> print = mapped.print();

        env.execute("kafka");
    }
}
